package com.greate.community.event;

import com.alibaba.fastjson.JSONObject;
import com.greate.community.entity.DiscussPost;
import com.greate.community.entity.Event;
import com.greate.community.entity.Message;
import com.greate.community.entity.UserBehavior;
import com.greate.community.mapper.DiscussPostMapper;
import com.greate.community.service.*;
import com.greate.community.util.CommunityConstant;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.ArrayList;

/**
 * 事件消费者
 */
@Component
public class EventConsumer implements CommunityConstant {

    private static final Logger logger = LoggerFactory.getLogger(EventConsumer.class);

    @Autowired
    private MessageService messageService;

    @Autowired
    private DiscussPostService discussPostService;

    @Autowired
    private ElasticsearchService elasticsearchService;

    @Autowired
    private UserBehaviorService userBehaviorService;

    @Autowired
    private RecommendCacheService recommendCacheService;

    @Autowired
    private DiscussPostMapper discussPostMapper;

    /**
     * 消费评论、点赞、关注事件
     * @param record
     */
    @KafkaListener(topics = {TOPIC_COMMNET, TOPIC_LIKE, TOPIC_FOLLOW})
    public void handleMessage(ConsumerRecord record) {
        try {
            if (record == null || record.value() == null) {
                logger.error("消息的内容为空");
                return ;
            }
            Event event = JSONObject.parseObject(record.value().toString(), Event.class);
            if (event == null) {
                logger.error("消息格式错误");
                return ;
            }

            // 发送系统通知
            Message message = new Message();
            message.setFromId(SYSTEM_USER_ID);
            message.setToId(event.getEntityUserId());
            message.setConversationId(event.getTopic());
            message.setCreateTime(new Date());

            Map<String, Object> content = new HashMap<>();
            content.put("userId", event.getUserId());
            content.put("entityType", event.getEntityType());
            content.put("entityId", event.getEntityId());
            if (!event.getData().isEmpty()) { // 存储 Event 中的 Data
                for (Map.Entry<String, Object> entry : event.getData().entrySet()) {
                    content.put(entry.getKey(), entry.getValue());
                }
            }
            message.setContent(JSONObject.toJSONString(content));

            messageService.addMessage(message);
        } catch (Exception e) {
            logger.error("消费Kafka消息异常", e);
        }
    }

    /**
     * 消费发帖事件
     */
    @KafkaListener(topics = {TOPIC_PUBLISH})
    public void handlePublishMessage(ConsumerRecord record) {
        try {
            if (record == null || record.value() == null) {
                logger.error("消息的内容为空");
                return ;
            }
            Event event = JSONObject.parseObject(record.value().toString(), Event.class);
            if (event == null) {
                logger.error("消息格式错误");
                return ;
            }

            DiscussPost post = discussPostService.findDiscussPostById(event.getEntityId());
            elasticsearchService.saveDiscusspost(post);
        } catch (Exception e) {
            logger.error("消费Kafka消息异常", e);
        }
    }

    /**
     * 消费删帖事件
     */
    @KafkaListener(topics = {TOPIC_DELETE})
    public void handleDeleteMessage(ConsumerRecord record) {
        try {
            if (record == null || record.value() == null) {
                logger.error("消息的内容为空");
                return ;
            }
            Event event = JSONObject.parseObject(record.value().toString(), Event.class);
            if (event == null) {
                logger.error("消息格式错误");
                return ;
            }

            elasticsearchService.deleteDiscusspost(event.getEntityId());
        } catch (Exception e) {
            logger.error("消费Kafka消息异常", e);
        }
    }

    @KafkaListener(topics = "behavior-topic")
    public void handleBehavior(ConsumerRecord<String, String> record) {
        if (record == null || record.value() == null) return;
        try {
            UserBehavior behavior = JSONObject.parseObject(record.value(), UserBehavior.class);
            // 持久化到MySQL
            userBehaviorService.saveBehavior(behavior);
        } catch (Exception e) {
            // 日志记录
        }
    }

    public List<DiscussPost> recommendPosts(int userId, int offset, int limit) {
        List<Integer> postIds = recommendCacheService.getRecommendPosts(userId);
        if (postIds != null && !postIds.isEmpty()) {
            // 只取分页部分
            List<Integer> pageIds = postIds.subList(offset, Math.min(offset + limit, postIds.size()));
            // 批量查详情
            return discussPostService.findPostsByIds(pageIds);
        }
        // 否则走原有推荐逻辑
        return null;
    }

    public List<DiscussPost> findPostsByIds(List<Integer> ids) {
        if (ids == null || ids.isEmpty()) return new ArrayList<>();
        return discussPostMapper.selectPostsByIds(ids);
    }

}
